#include "muduo/net/TcpServer.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/net/EventLoop.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"     //不加incllude可以吗
#include "muduo/base/Mutex.h"

#include "../mqcommon/logger.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/request.pb.h"
#include "../mqcommon/threadpool.hpp"

#include "virtualhost.hpp"
#include "consumer.hpp"
#include "connection.hpp"
#include <functional>

using namespace std::placeholders;
class Server
{
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
#define DBFILE "data.db"
public:
    Server(uint16_t port, const std::string& basedir)
        :_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "Server", muduo::net::TcpServer::kReusePort)
        ,_dispatcher(std::bind(&Server::onUnknownMessage, this, _1, _2, _3))
        ,_codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, _1, _2,_3)))
        ,_host(std::make_shared<VirtualHost>("host1", basedir, basedir + "/" + DBFILE))
        ,_consumerManeger(std::make_shared<QueueConsumerManager>())
        ,_connectionManager(std::make_shared<connectionManager>())
        ,_threadpool(std::make_shared<ThreadPool>())
    {
        std::unordered_map<std::string, MsgQueue::Ptr> queuemp = _host->getAllMsgQueue();
        for(auto& queue : queuemp)
            _consumerManeger->initQueueConsumer(queue.first);           //加载队列的所有消费者

        _dispatcher.registerMessageCallback<openChannelRequest>(std::bind(&Server::onOpenChannelRequestCallback, this, _1, _2, _3));
        _dispatcher.registerMessageCallback<closeChannelRequest>(std::bind(&Server::onCloseChannelRequestCallback, this, _1, _2, _3));

        _dispatcher.registerMessageCallback<declareExchangeRequest>(std::bind(&Server::onDeclareExchangeRequestCallback, this, _1, _2, _3));
        _dispatcher.registerMessageCallback<deleteExchangeRequest>(std::bind(&Server::onDeleteExchangeRequestCallback, this, _1, _2, _3));

        _dispatcher.registerMessageCallback<declareMsgQueueRequest>(std::bind(&Server::onDeclareMsgQueueRequestCallback, this, _1, _2, _3));
        _dispatcher.registerMessageCallback<deleteMsgQueueRequest>(std::bind(&Server::onDeleteMsgQueueRequestCallback, this, _1, _2, _3));

        _dispatcher.registerMessageCallback<bindRequest>(std::bind(&Server::onBindRequestCallback, this, _1, _2, _3));
        _dispatcher.registerMessageCallback<unBindRequest>(std::bind(&Server::onUnBindRequestCallback, this, _1, _2, _3));

        _dispatcher.registerMessageCallback<basicPublishRequest>(std::bind(&Server::onBasicPublishRequestCallback, this, _1, _2, _3));
        _dispatcher.registerMessageCallback<basicAckRequest>(std::bind(&Server::onBasicAckRequestCallback, this, _1, _2, _3));
        _dispatcher.registerMessageCallback<basicConsumerRequest>(std::bind(&Server::onBasicConsumerRequestCallback, this, _1, _2, _3));
        _dispatcher.registerMessageCallback<basicCancelRequest>(std::bind(&Server::onBasicCancelRequestCallback, this, _1, _2, _3));

        _dispatcher.registerMessageCallback<pullOnceRequest>(std::bind(&Server::onPullOnceRequestCallback, this, _1, _2, _3));
        
        _server.setConnectionCallback(std::bind(&Server::OnConnectionCallback, this, _1));
        _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), _1, _2, _3));
    }
    void Start()
    {
        _server.start();
        _baseloop.loop();
    }
private:
    //打开关闭信道
    void onOpenChannelRequestCallback(const muduo::net::TcpConnectionPtr& conn, const openChannelRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr connectionPtr = _connectionManager->getConnection(conn);
        if(connectionPtr.get() == nullptr)
        {
            DLOG("连接获取失败，conn没有对应的connection对象");
            return;
        }
        connectionPtr->openChannel(message);
    }
    void onCloseChannelRequestCallback(const muduo::net::TcpConnectionPtr& conn, const closeChannelRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("关闭信道失败，conn没有对应的connection对象");
            return;
        }
        selfconn->closeChannel(message);
    }
    //声明删除交换机
    void onDeclareExchangeRequestCallback(const muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("声明交换机失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("声明交换机失败, 链接没有对应的信道");
            return;
        }
        channelPtr->declareExchange(message);
    }
    void onDeleteExchangeRequestCallback(const muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("删除交换机失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("删除交换机失败, 链接没有对应的信道");
            return;
        }
        channelPtr->deleteExchange(message);
    }
    //声明删除队列
    void onDeclareMsgQueueRequestCallback(const muduo::net::TcpConnectionPtr& conn, const declareMsgQueueRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("声明队列失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("声明队列失败, 链接没有对应的信道");
            return;
        }
        channelPtr->declareMsgQueue(message);
    }
    void onDeleteMsgQueueRequestCallback(const muduo::net::TcpConnectionPtr& conn, const deleteMsgQueueRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("删除队列失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("删除队列失败, 链接没有对应的信道");
            return;
        }
        channelPtr->deleteMsgQueue(message);
    }
    //绑定解除绑定
    void onBindRequestCallback(const muduo::net::TcpConnectionPtr& conn, const bindRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("绑定失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("绑定失败, 链接没有对应的信道");
            return;
        }
        channelPtr->bind(message);
    }
    void onUnBindRequestCallback(const muduo::net::TcpConnectionPtr& conn, const unBindRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("解除绑定失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("解除绑定失败, 链接没有对应的信道");
            return;
        }
        channelPtr->unBind(message);
    }
    //发布消息
    void onBasicPublishRequestCallback(const muduo::net::TcpConnectionPtr& conn, const basicPublishRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("发布消息失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("发布消息失败, 链接没有对应的信道");
            return;
        }
        DLOG("%s", message->body().c_str());
        channelPtr->basicPublish(message);
    }
    //响应消息
    void onBasicAckRequestCallback(const muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("响应消息失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("响应消息失败, 链接没有对应的信道");
            return;
        }
        channelPtr->basicAck(message);
    }
    //订阅消息
    void onBasicConsumerRequestCallback(const muduo::net::TcpConnectionPtr& conn, const basicConsumerRequestPtr& message, muduo::Timestamp)
    {
        DLOG("消费者");
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("订阅消息失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("订阅消息失败, 链接没有对应的信道");
            return;
        }
        channelPtr->basicConsumer(message);
    }
    //取消订阅
    void onBasicCancelRequestCallback(const muduo::net::TcpConnectionPtr& conn, const basicCancelRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("取消订阅失败, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("取消订阅失败, 链接没有对应的信道");
            return;
        }
        channelPtr->basicCancel(message);
    }
    //创建消费者的时候拉取一次
    void onPullOnceRequestCallback(const muduo::net::TcpConnectionPtr& conn, const pullOnceRequestPtr& message, muduo::Timestamp)
    {
        connection::Ptr selfconn = _connectionManager->getConnection(conn);
        if(selfconn.get() == nullptr)
        {
            DLOG("拉取, conn没有对应的connection对象");
            return;
        }
        channel::Ptr channelPtr = selfconn->getChannel(message->cid());
        if(channelPtr.get() == nullptr)
        {
            DLOG("拉取, 链接没有对应的信道");
            return;
        }
        channelPtr->pullOnce(message);
    }


    void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,
                        const MessagePtr& message,
                        muduo::Timestamp)
    {
        LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
        conn->shutdown();
    }
    void OnConnectionCallback(const muduo::net::TcpConnectionPtr& conn)
    {
        if(conn->connected())
            _connectionManager->createConnetion(conn, _codec, _consumerManeger, _host, _threadpool);        //链接建立，马上创建connection对象
        else std::cout << "链接断开" << std::endl;
    }
private:
    muduo::net::EventLoop _baseloop;
    muduo::net::TcpServer _server;
    ProtobufDispatcher _dispatcher;
    ProtobufCodecPtr _codec;

    VirtualHost::Ptr _host;
    QueueConsumerManager::Ptr _consumerManeger;
    connectionManager::Ptr _connectionManager;
    ThreadPool::Ptr _threadpool;
};